﻿/*----------------------------------------------------------------------------------------
*
*  Copyright 2019, Gao Hai Hui, <fromzeropoint@126.com>.  All rights reserved.
*  https://gitee.com/helloworldghh/xoskit.git
*  Use of this source code is governed by a MIT license
*  that can be found in the License file.
*
----------------------------------------------------------------------------------------*/
#include "../import/head.h"
#include "../global/head.h"
#include "../macro/head.h"
#include "aio_epoll_notify.h"
#include "aio_helper.h"
#include "aio_thread.h"
#include "aio_impl.h"

/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos_aio
{
    aio_thread::aio_thread( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
        m_bQuit = false;
        m_pThread = 0;
    }

    aio_thread::~aio_thread()
    {
    }

    ////////////////////////////////////////////////////////////////////////////////////
    // 

    int aio_thread::init()
    {
        int nRet = 0;

        if( 0 == nRet )
        {
            nRet = g_pSysData->m_pContainer->xos()->create(xos::i_xos::XOS_OBJ_THREAD, ( void ** )&m_pThread);
        }

        if( 0 == nRet )
        {
            m_pThread->create( aio_thread::thread, this );
        }

        return nRet;
    }

    int aio_thread::term()
    {
        int nRet = 0;

        if( m_pThread )
        {
            m_bQuit = true;
            m_pAioImpl->m_pEpollNotify->set_event();
            m_pThread->wait();
            m_pThread->release();
            m_pThread = 0;
        }

        m_pAioImpl = 0;

        return nRet;
    }

    //////////////////////////////////////////////////////////////////////////////////////////////
    // 

    void * aio_thread::thread( void * lpData )
    {
        using namespace std;

        aio_thread & obj = *( aio_thread * )lpData;
        aio_impl * lpAioImpl = obj.m_pAioImpl;

        while( !obj.m_bQuit )
        {
            int nNum = kevent( lpAioImpl->m_hKqueue, NULL, 0, obj.m_event, sizeof( obj.m_event ) / sizeof( obj.m_event[0] ), 0 );
            if( nNum > 0 )
            {
                obj.run( nNum ); 
            }
        }

        return 0;
    }

    int aio_thread::run( int nNum )
    {
        using namespace std;
        int nRet = 0;

        aio_data::LIST & data_list = m_pAioImpl->m_return_list;

        if( 0 == nRet )
        {
            data_list.put_back_to_pool( true );
        }

        for( int i = 0; ( 0 == nRet ) && ( i < nNum ); ++i )
        {
            helper_run( m_event[i] );
        }

        for( aio_data::ITER i = data_list.begin(); ( 0 == nRet ) && ( i != data_list.end() ); ++i )
        {
            aio_data * p = *i;
            xos_box::i_msg * pMsg = p->m_pMsg;
            aio_key * pKey = (aio_key *)pMsg->get_void( 1, 0 );
            pKey->finish_request( m_pAioImpl );
            if( aio_key::KEY_STATE_RUNNING == pKey->m_eState )
            {
                pMsg->set_void( 1, 0 );
                pKey->release();
                pKey = 0;
            }
            m_pAioImpl->notify( pMsg );
            p->m_pMsg = 0;
            pMsg = 0;
        }

        data_list.put_back_to_pool( true );

        return nRet;
    }

    int aio_thread::helper_run( struct kevent & e )
    {
        using namespace std;
        int nRet = 0;

        aio_helper tool( m_pAioImpl );

        void * lpData = e.udata;
        int nFilter = e.filter;
        int nFlags = e.flags;
        int nData = e.data;

        if( lpData != m_pAioImpl->m_pEpollNotify )
        {
            aio_key * pAioData = ( aio_key * )lpData;
            tool.epoll_proc( pAioData, nFilter, nFlags, nData );
        }
        else
        {
            m_pAioImpl->m_pEpollNotify->reset();
            tool.user_request();
        }

        return nRet;
    }

} // xos_aio
